package com.atguigui.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.SourceTransactionBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

public class DwsSourceTransactionWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsSourceTransactionWindow().start(
                12326,
                4,
                "dws_source_source_transaction_window",
                Constant.DWD_TRANS_PAY
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //kafkaStrDS.print();
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSON.parseObject(jsonStr);
                    }
                }
        );

        //以唯一主键分组去重
        KeyedStream<JSONObject, String> idKeyedDs = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //定时器去重 只获取最后一个
        SingleOutputStreamOperator<JSONObject> distinctDs = idKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> jsonObjectValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<JSONObject> stateProperties =
                                new ValueStateDescriptor<JSONObject>("jsonObjectValueState", JSONObject.class);
                        jsonObjectValueState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        JSONObject lastJsonObj = jsonObjectValueState.value();
                        if (lastJsonObj == null) {
                            jsonObjectValueState.update(jsonObj);

                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = jsonObjectValueState.value();
                        out.collect(jsonObj);
                    }
                }
        );

        //distinctDs.print();

        KeyedStream<JSONObject, String> useridKeyedDs = distinctDs.keyBy(jsonObj -> jsonObj.getString("user_id"));

        SingleOutputStreamOperator<JSONObject> processDs = useridKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> lastUserIdState;
                    ValueState<String> lastOrderIdState;


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateProperties =
                                new ValueStateDescriptor<String>("lastUserIdState", String.class);
                        lastUserIdState = getRuntimeContext().getState(stateProperties);

                        ValueStateDescriptor<String> lastOrderIdStateDesc =
                                new ValueStateDescriptor<String>("lastOrderIdState", String.class);
                        lastOrderIdState = getRuntimeContext().getState(lastOrderIdStateDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = new JSONObject();
                        BigDecimal amount = jsonObj.getBigDecimal("final_amount");
                        String sc = jsonObj.getString("sc");
                        String userId = jsonObj.getString("user_id");
                        String orderId = jsonObj.getString("order_id");
                        Long ts = jsonObj.getLong("ts") * 1000  ;

                        String lastUserId = lastUserIdState.value();
                        String lastOrderId = lastOrderIdState.value();

                        // 第一次
                        jsonObject.put("sc",sc);
                        jsonObject.put("amount", amount); //金额
                        jsonObject.put("orderUserId", 1); //下单人数
                        jsonObject.put("orderCount", 1); //下单次数
                        jsonObject.put("uuCt", 1); //独立用户
                        jsonObject.put("ts",ts);


                        if (lastUserId != null) {
                            jsonObject.put("uuCt", 0); //独立用户
                        }
                        if (lastOrderId != null && lastOrderId.equals(orderId)) { //如果order_id相同算同一次下单
                            jsonObject.put("orderCount", 0); //下单次数
                        }

                        lastUserIdState.update(userId);
                        lastOrderIdState.update(orderId);
                        out.collect(jsonObject);
                    }
                }
        );

        //processDs.print();
        //4> {"sc":"1","amount":200.0,"orderUserId":1,"uuCt":1,"orderCount":1}
        //4> {"sc":"2","amount":200.0,"orderUserId":1,"uuCt":1,"orderCount":1}
        //1> {"sc":"2","amount":200.0,"orderUserId":1,"uuCt":1,"orderCount":1}

        SingleOutputStreamOperator<SourceTransactionBean> stBean = processDs.map(
                new MapFunction<JSONObject, SourceTransactionBean>() {
                    @Override
                    public SourceTransactionBean map(JSONObject value) throws Exception {
                        SourceTransactionBean sourceTransactionBean = new SourceTransactionBean();
                        sourceTransactionBean.setStt("");
                        sourceTransactionBean.setEdt("");
                        sourceTransactionBean.setCur_date("");
                        sourceTransactionBean.setSc(value.getString("sc"));
                        sourceTransactionBean.setSumAmount(value.getDouble("amount"));
                        sourceTransactionBean.setOrderUserSum(value.getLong("orderUserId"));
                        sourceTransactionBean.setOrderCount(value.getLong("orderCount"));
                        sourceTransactionBean.setZhl(0.0);
                        sourceTransactionBean.setUuCt(value.getLong("uuCt"));
                        sourceTransactionBean.setTs(value.getLong("ts"));
                        return sourceTransactionBean;
                    }
                }
        );

        //stBean.print();

        SingleOutputStreamOperator<SourceTransactionBean> withWatermarkDs = stBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<SourceTransactionBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<SourceTransactionBean>() {
                                    @Override
                                    public long extractTimestamp(SourceTransactionBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        //1> {"sc":"3","amount":200.0,"orderUserId":1,"uuCt":1,"orderCount":1,"ts":1731682216000}
        KeyedStream<SourceTransactionBean, String> scKeyedDs = withWatermarkDs.keyBy(sourceTransactionBean -> sourceTransactionBean.getSc());

        //scKeyedDs.print();
        //SourceTransactionBean(stt=, edt=, cur_date=, scId=1, sumAmount=200.0, orderUserSum=null, orderCount=1, zhl=0.0, uuCt=1, ts=1731683499000)
        WindowedStream<SourceTransactionBean, String, TimeWindow> windowDs = scKeyedDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));


        SingleOutputStreamOperator<SourceTransactionBean> reduceDs = windowDs.reduce(
   // SourceTransactionBean(stt=, edt=, cur_date=, scId=1, sumAmount=200.0, orderUserSum=1, orderCount=1, zhl=0.0, uuCt=1, ts=1731683766000)
                new ReduceFunction<SourceTransactionBean>() {
                    @Override
                    public SourceTransactionBean reduce(SourceTransactionBean value1, SourceTransactionBean value2) throws Exception {
                        value1.setSumAmount(value1.getSumAmount() + value2.getSumAmount());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderUserSum(value1.getOrderUserSum() + value2.getOrderUserSum());
                        value1.setUuCt(value1.getUuCt() + value2.getUuCt());
                        return value1;
                    }
                },
                new ProcessWindowFunction<SourceTransactionBean, SourceTransactionBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<SourceTransactionBean, SourceTransactionBean, String, TimeWindow>.Context context, Iterable<SourceTransactionBean> elements, Collector<SourceTransactionBean> out) throws Exception {
                        SourceTransactionBean stb = elements.iterator().next();
                        TimeWindow window = context.window();
                        SourceTransactionBean sourceTransactionBean = new SourceTransactionBean();
                        sourceTransactionBean.setStt(DateFormatUtil.tsToDateTime(window.getStart()));
                        sourceTransactionBean.setEdt(DateFormatUtil.tsToDateTime(window.getEnd()));
                        sourceTransactionBean.setCur_date(DateFormatUtil.tsToDate(window.getStart()));
                        sourceTransactionBean.setSc(stb.getSc());
                        sourceTransactionBean.setSumAmount(stb.getSumAmount());
                        sourceTransactionBean.setOrderUserSum(stb.getOrderUserSum());
                        sourceTransactionBean.setOrderCount(stb.getOrderCount());
                        sourceTransactionBean.setZhl((double) ( (double)stb.getUuCt() / (double)stb.getOrderUserSum()));
                        sourceTransactionBean.setUuCt(stb.getUuCt());
                        sourceTransactionBean.setTs(stb.getTs());
                        out.collect(sourceTransactionBean);
                    }
                }
        );

        //reduceDs.print();

        SingleOutputStreamOperator<SourceTransactionBean> sourcesiteDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<SourceTransactionBean>() {
                    @Override
                    protected void addDim(SourceTransactionBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSc(dimJsonObj.getString("source_site"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_source";
                    }

                    @Override
                    public String getRowKey(SourceTransactionBean orderBean) {
                        return orderBean.getSc();
                    }
                }
                ,
                60, TimeUnit.SECONDS
        );

        //sourcesiteDs.print();
        SingleOutputStreamOperator<String> beanDs = sourcesiteDs.map(new BeanToJsonStrMapFunction<>());
        beanDs.print();

        beanDs.sinkTo(FlinkSinkUtil.getDorisSink("dws_source_source_transaction_window"));
    }
}
